package RDD练习

import org.apache.spark.{SparkConf, SparkContext}

/**
  * 12 宋江 25 男 chinese 50
  * 班级 姓名 年龄 性别 国籍 分数
  */
object RDDTest {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Test").setMaster("local[6]")
    val sc = new SparkContext(conf)
    val source = sc.textFile("E:/IDEA/IntelliJ IDEA 2019.1/sp/spark练习项目/src/main/java/RDD练习/data")
    //一共有多少个小于20岁的人参加考试？
    val counts1 = source.groupBy {
      x =>
        val datas = x.split(" ")
        datas(0) + " " + datas(1) + " " + datas(2)
    }.filter(_._1.split(" ")(2).toInt < 20).count()
    //一共有多个男生参加考试？
    val man_count = source.groupBy {
      x =>
        val datas = x.split(" ")
        datas(0) + " " + datas(1) + " " + datas(3)
    }.filter(_._1.split(" ")(2).toString.equals("男")).count()
    //12班有多少人参加考试？
    val person_count = source.groupBy {
      x =>
        val datas = x.split(" ")
        datas(0) + " " + datas(1) + " " + datas(2)
    }.filter(_._1.split(" ")(0).toInt == 12).count()
    //语文科目的平均成绩是多少？
    val chinese_avg = source.filter(
      _.split(" ")(4).toString.equals("chinese")
    ).map(_.split(" ")(5).toFloat).mean()
    //单个人平均成绩是多少？
    val person_avg = source.map {
      item =>
        val datas = item.split(" ")
        (datas(0) + "," + datas(1), datas(5).toInt)
    }.map(x => (x._1, (x._2, 1))).reduceByKey((n, m) => (n._1 + m._1, m._2 + n._2)).map(x => (x._1, x._2._1 / x._2._2))
    //12班平均成绩是多少？
    val class_avg = source.map {
      x =>
        val datas = x.split(" ")
        (datas(0) + "," + datas(4), datas(5).toInt)
    }.filter(_._1.split(",")(0).toInt == 12)
      .map(x => (x._1, (x._2, 1))).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))
      .map(x => (x._1, x._2._1 / x._2._2))
    //12班男生平均总成绩是多少？
    val class_woman_avg = source.filter({
      x =>
        val datas = x.split(" ")
        datas(0).toInt == 12 & datas(3).equals("男")
    }).map {
      x =>
        val datas = x.split(" ")
        (datas(0), datas(5).toInt)
    }.map(x => (x._1, (x._2, 1))).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2)).mapValues(x => x._1 / x._2).values
    //全校语文成绩最高分是多少？
    val chinese_max = source.filter {
      _.split(" ")(4).equals("chinese")
    }
      .map(_.split(" ")(5)).max()
    //总成绩大于150分的12班的女生有几个？
    val sum_woman = source.map {
      x =>
        val datas = x.split(" ")
        ((datas(1), datas(3)), datas(5).toInt)
    }.reduceByKey((curr, agg) => curr + agg)
      .filter {
        x =>
          x._1._2.equals("女") && x._2 > 150
      }.count()
    println("年龄小于20的人数是:" + counts1)
    println("男生的人数:" + man_count)
    println("12班考试人数:" + person_count)
    println("语文的平均成绩:" + chinese_avg)
    println("单个人的平均成绩:")
    person_avg.foreach(println(_))
    println("12班的平均成绩:")
    class_avg.foreach(println(_))
    println("12班男的平均成绩:")
    class_woman_avg.foreach(println(_))
    println("语文成绩的最高分:" + chinese_max)
    println("总成绩大于150的女生人数:"+sum_woman)
  }
}
